package com.richard.library.simplerx.concurrent;

import com.richard.library.simplerx.XObservableOnSubscribe;
import com.richard.library.simplerx.XObservable;

import java.util.List;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;


/**
 * <pre>
 * Description : 并发执行观察者
 * Author : admin-richard
 * Date : 2020/3/4 13:48
 * Changelog:
 * Version            Date            Author              Detail
 * ----------------------------------------------------------------------
 * 1.0         2020/3/4 13:48     admin-richard         new file.
 * </pre>
 */
public final class XObservableConcurrent {

    private final XObservableOnSubscribe<List<ExecuteTask>> xObservableOnSubscribe;

    private XObservableConcurrent(XObservableOnSubscribe<List<ExecuteTask>> xObservableOnSubscribe) {
        this.xObservableOnSubscribe = xObservableOnSubscribe;
    }

    public static <T> XObservableConcurrent create(XObservableOnSubscribe<List<ExecuteTask>> xObservableOnSubscribe) {
        return new XObservableConcurrent(xObservableOnSubscribe);
    }

    /**
     * 多线程并发执行任务
     * 处理线程：子(异步)线程
     * 回调线程：由isForceCallbackToMainThread参数决定
     *
     * @param isSubTaskForceCallToUIThread 必填 子级任务执行完成之后是否强制回调至主线程，true:回调处为主线程、false:回调处线程由该方法调用处线程决定
     * @param onSubTaskFinishListener      选填 当某子任务执行完成时回调
     */
    public XObservable<List<ExecuteTask>> concurrentExecute(boolean isSubTaskForceCallToUIThread, OnSubTaskFinishListener onSubTaskFinishListener) {
        return XObservable.create(() -> {
            Observable<ExecuteTask> innerObservable = Observable
                    .fromIterable(xObservableOnSubscribe.run())
                    .flatMap((Function<ExecuteTask, ObservableSource<ExecuteTask>>) syncTask -> Observable
                            .just(syncTask)
                            .observeOn(Schedulers.io())
                            .map(executeTask -> {
                                syncTask.invokeRunMethod();
                                return syncTask;
                            }));

            if (isSubTaskForceCallToUIThread) {
                innerObservable = innerObservable.observeOn(AndroidSchedulers.mainThread());
            }

            return innerObservable
                    .map(executeTask -> {
                        if (onSubTaskFinishListener != null) {
                            onSubTaskFinishListener.onSubTaskFinish(executeTask);
                        }
                        return executeTask;
                    })
                    .toList()
                    .blockingGet();
        });
    }

}
